package test.sunyu.storm.demo.demo2;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.utils.Utils;
import org.slf4j.Logger;
import sunyu.kit.common.LogKit;

/**
 * 读取随机字符串的拓扑
 *
 * @author 孙宇
 */
public class TopologyMain {

    private static final Logger logger = LogKit.getLogger();

    public static void main(String[] args) {
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("randomSpout", new RandomSpout());
        //builder.setBolt("senqueceBolt", new SenqueceBolt()).shuffleGrouping("randomSpout");
        builder.setBolt("senqueceBolt", new SenqueceBolt()).localOrShuffleGrouping("randomSpout");
        Config conf = new Config();
        conf.setDebug(true);
        if (args != null && args.length > 0) {
            logger.info("正在提交拓扑到集群处理");
            conf.setNumWorkers(3);
            try {
                StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
            } catch (AlreadyAliveException e) {
                e.printStackTrace();
            } catch (InvalidTopologyException e) {
                e.printStackTrace();
            } catch (AuthorizationException e) {
                e.printStackTrace();
            }
        } else {
            logger.info("正在提交拓扑到本地处理");
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("randomStringTopo", conf, builder.createTopology());
            Utils.sleep(1000 * 20);
            cluster.killTopology("randomStringTopo");
            cluster.shutdown();
        }
    }

}
